
// BSD License (http://www.galagosearch.org/license)

package org.galagosearch.core.parse;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.FileInputStream;
import org.galagosearch.tupleflow.Counter;
import org.galagosearch.tupleflow.InputClass;
import org.galagosearch.tupleflow.OutputClass;
import org.galagosearch.tupleflow.StandardStep;
import org.galagosearch.tupleflow.execution.Verified;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.zip.GZIPInputStream;
import org.galagosearch.tupleflow.StreamCreator;
import org.galagosearch.tupleflow.TupleFlowParameters;
import org.galagosearch.core.types.DocumentSplit;
import org.galagosearch.core.types.VersionedDocumentSplit;
import org.galagosearch.tupleflow.Parameters;

/**
 *
 * @author trevor
 */
@Verified
@InputClass(className = "org.galagosearch.core.types.VersionedDocumentSplit")
@OutputClass(className = "org.galagosearch.core.parse.VersionedDocument")
public class VersionedUniversalParser extends StandardStep<VersionedDocumentSplit, VersionedDocument> {
    private Counter documentCounter;
    private Parameters parameters;
    private static int docID=0;
    public BufferedReader getBufferedReader(VersionedDocumentSplit split) throws IOException {
        FileInputStream stream = StreamCreator.realInputStream(split.fileName);
        BufferedReader reader;

        if (split.isCompressed) {
            reader = new BufferedReader(new InputStreamReader(new GZIPInputStream(stream)));
        } else {
            reader = new BufferedReader(new InputStreamReader(stream));
        }
        return reader;
    }
    
    public BufferedInputStream getBufferedInputStream(VersionedDocumentSplit split) throws IOException {
        FileInputStream fileStream = StreamCreator.realInputStream(split.fileName);
        BufferedInputStream stream;

        if (split.isCompressed) {
            stream = new BufferedInputStream(new GZIPInputStream(fileStream));
        } else {
            stream = new BufferedInputStream(fileStream);
        }
        return stream;
    }
    
    public VersionedUniversalParser(TupleFlowParameters parameters) {
        documentCounter = parameters.getCounter("Documents Parsed");
        this.parameters = parameters.getXML();
        //String DI = parameters.getXML().get("docID");
    }
    
    public void process(VersionedDocumentSplit split) throws IOException {
        VersionedDocumentStreamParser parser;

        if (split.fileType.equals("html") ||
            split.fileType.equals("xml") ||
            split.fileType.equals("txt")) {
        	if(parameters.containsKey("docID"))
        		parameters.set("docID", docID+"");
        	else parameters.add("docID", docID+"");
            docID = docID + 1;
            parser = new VersionedFileParser(parameters, split.fileName, getBufferedReader(split)); //we should reach here
       // } else if (split.fileType.equals("arc")) {
       //     parser = new ArcParser(getBufferedInputStream(split));
       // } else if (split.fileType.equals("trectext")) {
       //     parser = new TrecTextParser(getBufferedReader(split));
       // } else if (split.fileType.equals("trecweb")) {
       //     parser = new TrecWebParser(getBufferedReader(split));
       // } else if (split.fileType.equals("corpus")) {
       //     parser = new IndexReaderSplitParser(split);
        } else {
            throw new IOException("Unknown fileType: " + split.fileType +
                                  " for fileName: "  + split.fileName);
        }
        try{
        VersionedDocument[] documents;
        //System.out.println(docID+"");
        while ((documents = parser.nextDocument()) != null) {
        	for(int i=0;i<documents.length;i++){
        		VersionedDocument document = documents[i];
        		//document.metadata.put("docID", parameters.get("docID","-1"));
        		//document.metadata.put("docID", ""+docID);
        		//System.out.println(document.metadata.get("docID")+"."+i+"\t"+document.metadata.get("title")+"\t"+document.metadata.get("timeBegin")+"");
        		processor.process(document);
                if (documentCounter != null)
                    documentCounter.increment();
        	}  
        }
        }catch(Exception e){e.printStackTrace();}
    }
}
